package disruption

import (
	"bufio"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/emirpasic/gods/sets/hashset"
)

/**
 * @description: 读取学科的linksout 文件, deprecated
 * @param {*}
 * @return subjectSet 学科文章集合
 * @return subjectOutSet 学科out的文章集合
 */
func ReadLinsout(subjectFile string) (subjectSet, subjectOutSet *hashset.Set) {
	subjectOutSet = hashset.New()
	subjectSet = hashset.New()

	fi, err := os.Open("/home/ni/data/mag/reference/delete_noref_v2/" + subjectFile)
	if err != nil {
		fmt.Printf("Error: %s\n", err)
		return
	}
	defer fi.Close()

	start_time := time.Now()
	br := bufio.NewReader(fi)
	for {
		a, c := br.ReadString('\n')
		if c == io.EOF {
			break
		}
		strs := strings.Split(string(a), "\t")

		subjectOutSet.Add(strs[2])
		subjectSet.Add(strs[0])
	}
	end_time := time.Now()
	fmt.Println("读学科耗时", end_time.Sub(start_time))
	return
}

/**
 * @description: 统计所有集合的节点情况，全部加载到内存，空间换时间
 * @return {*}
 */
func ReadAllLinks() (allRef *map[string]*[]string) {
	allRef = new(map[string]*[]string)
	*allRef = make(map[string]*[]string)
	// *allRef = map[string]*[]string

	fi, err := os.Open("/home/ni/data/mag/all_refed_authors_data.txt")
	// fi, err := os.Open("test/ta.txt")
	if err != nil {
		fmt.Printf("Error: %s\n", err)
		return
	}
	defer fi.Close()
	start_time := time.Now()
	// br := bufio.NewReaderSize(fi, 50000)
	br := bufio.NewReader(fi)

	for {
		a, c := br.ReadString('\n')
		if c == io.EOF {
			break
		}
		strs := strings.Split(a, "\t")
		linksoutArray := []string{}
		if len(strs) >= 3 {
			err := json.Unmarshal([]byte(strs[2]), &linksoutArray)
			if err != nil {
				fmt.Printf("Error: %s\n", err)
				fmt.Printf("%+v\n", string(a))
				// a, _, c = br.ReadLine()
				// fmt.Printf("%+v\n", string(a))
				return
			}
		} else {
			fmt.Println("strs", strs)
		}
		if len(linksoutArray) > 0 {
			(*allRef)[strs[0]] = &linksoutArray
		}
	}
	end_time := time.Now()
	fmt.Println("read all links:", end_time.Sub(start_time))
	return
}

type ChanObj struct {
	Key string
	S   *[]string
}

/**
 * @description: 统计所有集合的节点情况，全部加载到内存，空间换时间
 * @return {*}
 */
func ReadAllLinksChan() (allRef *map[string]*[]string) {
	var wg = sync.WaitGroup{}
	var wgOut = sync.WaitGroup{}
	allRef = new(map[string]*[]string)
	*allRef = make(map[string]*[]string)
	// *allRef = map[string]*[]string

	fi, err := os.Open("/home/ni/data/mag/all_refed_authors_data.txt")
	// fi, err := os.Open("test/ta.txt")
	if err != nil {
		fmt.Printf("Error: %s\n", err)
		return
	}
	defer fi.Close()
	start_time := time.Now()
	// br := bufio.NewReaderSize(fi, 50000)
	br := bufio.NewReader(fi)

	ch := make(chan string, 1000)
	chOut := make(chan ChanObj, 1000)
	const threaCount = 20
	wg.Add(threaCount)
	wgOut.Add(1)
	for i := 0; i < threaCount; i++ {
		go handleAllLinksChan(ch, chOut, &wg)
	}
	// 保存到字典
	go func() {
		for chobj := range chOut {
			(*allRef)[chobj.Key] = chobj.S
		}
		wgOut.Done()
	}()

	for {
		a, c := br.ReadString('\n')
		if c == io.EOF {
			break
		}
		ch <- a
	}
	close(ch)
	wg.Wait()
	close(chOut)
	wgOut.Wait()
	end_time := time.Now()
	fmt.Println("read all links:", end_time.Sub(start_time))
	return
}

func handleAllLinksChan(ch chan string, chOut chan ChanObj, wg *sync.WaitGroup) {
	for a := range ch {
		strs := strings.Split(a, "\t")
		linksoutArray := []string{}
		if len(strs) >= 3 {
			err := json.Unmarshal([]byte(strs[2]), &linksoutArray)
			if err != nil {
				fmt.Printf("Error: %s\n", err)
				fmt.Printf("%+v\n", string(a))
				return
			}
		} else {
			fmt.Println("strs", strs)
		}
		if len(linksoutArray) > 0 {
			co := ChanObj{Key: strs[0], S: &linksoutArray}
			chOut <- co
		}
	}
	wg.Done()
}

/**
 * @description: 计算处理程序，使用map，单线程计算
 * @param {*}
 * @return {*}
 */
func HandleChan(allRef *map[string]*[]string, subjectSet, subjectOutSet *hashset.Set) {

	var (
		ni, nj, nk int
	)

	start_time := time.Now()
	for ks, va := range *allRef {

		// 在 subject 内，排除计算
		if subjectSet.Contains(ks) {
			continue
		}

		flagNi := false
		flagNk := false

		for _, id := range *va {
			if subjectSet.Contains(id) {
				flagNi = true
				break
			}
		}

		for _, id := range *va {
			if subjectOutSet.Contains(id) {
				flagNk = true
				break
			}
		}

		if flagNi == true && flagNk == true {
			nj += 1
		} else if flagNk == true {
			nk += 1
		} else if flagNi == true {
			ni += 1
		}
	}

	end_time := time.Now()
	fmt.Println(end_time.Sub(start_time))
	fmt.Println(ni, nj, nk)
	fmt.Println((float64(ni) - float64(nj)) / (float64(ni) + float64(nj) + float64(nk)))
}

type RetChan struct {
	Ks string
	Va *[]string
}

/**
 * @description: 协程加速处理计算处理程序
 */
func HandleChanThead(allRef *map[string]*[]string, subjectSet, subjectOutSet *hashset.Set) (ni, nj, nk int64) {
	var wg = sync.WaitGroup{}

	ch := make(chan RetChan, 1000)
	start_time := time.Now()

	const threadCount = 10
	wg.Add(threadCount)
	for i := 0; i < threadCount; i++ {
		go HandleChanTheadAtom(ch, subjectSet, subjectOutSet, &ni, &nj, &nk, &wg)
	}

	for ks, va := range *allRef {
		rc := RetChan{Ks: ks, Va: va}
		ch <- rc
	}
	close(ch)
	wg.Wait()
	// <-time.After(2 * time.Second)
	end_time := time.Now()
	fmt.Println(end_time.Sub(start_time))
	fmt.Println(ni, nj, nk)
	fmt.Println((float64(ni) - float64(nj)) / (float64(ni) + float64(nj) + float64(nk)))
	return
}

/**
 * @description: 原子处理程序
 */
func HandleChanTheadAtom(ch chan RetChan, subjectSet, subjectOutSet *hashset.Set, ni, nj, nk *int64, wg *sync.WaitGroup) {
	var (
		nni, nnj, nnk int64
	)
	bk := 0
	for rc := range ch {
		// 在 subject 内，排除计算
		if subjectSet.Contains(rc.Ks) {
			bk += 1
			continue
		}

		flagNi := false
		flagNk := false

		for _, id := range *rc.Va {
			if subjectSet.Contains(id) {
				flagNi = true
				break
			}
		}

		for _, id := range *rc.Va {
			if subjectOutSet.Contains(id) {
				flagNk = true
				break
			}
		}

		if flagNi == true && flagNk == true {
			nnj += 1
		} else if flagNk == true {
			nnk += 1
		} else if flagNi == true {
			nni += 1
		}
	}
	// 同步计算滞后
	atomic.AddInt64(nj, nnj)
	atomic.AddInt64(nk, nnk)
	atomic.AddInt64(ni, nni)
	wg.Done()
}

/**
 * @description:处理测试程序
 */
func HandalMain() {
	allRef := ReadAllLinksChan()
	var (
		subjectSet, subjectOutSet *hashset.Set
	)
	AllSubjects := []string{"Number theory.txt", "Logic.txt", "Genome editing.txt", "Quantum computing.txt", "Experimental physics.txt", "Deep learning.txt", "Operating system.txt", "Genetic engineering.txt", "Literature.txt", "Geometry.txt", "Industrial engineering.txt", "Applied physics.txt", "Applied mathematics.txt", "Algebra.txt", "Cognitive science.txt", "Theoretical physics.txt", "Philosophy.txt", "Linguistics.txt", "Biological engineering.txt", "Anthropology.txt", "Computer engineering.txt", "Civil engineering.txt", "History.txt", "Theoretical computer science.txt", "Discrete mathematics.txt", "Political science.txt", "Environmental engineering.txt", "Mechanical engineering.txt", "Earth science.txt", "Geography.txt", "Machine learning.txt", "Electrical engineering.txt", "Chemical engineering.txt", "Environmental science.txt", "Sociology.txt", "Neuroscience.txt", "Artificial intelligence.txt", "Geology.txt", "Economics.txt", "Mathematics.txt", "Physics.txt", "Psychology.txt", "Materials science.txt", "Computer science.txt", "Engineering disciplines.txt", "Chemistry.txt", "Biology.txt", "Medicine.txt"}

	for _, subjectName := range AllSubjects {
		subjectSet, subjectOutSet = ReadLinsout(subjectName)
		ni, nj, nk := HandleChanThead(allRef, subjectSet, subjectOutSet)
		csvLine := []string{subjectName, strconv.FormatInt(ni, 10), strconv.FormatInt(nj, 10), strconv.FormatInt(nk, 10)}
		WriteCsvLine("all_subject.csv", &csvLine)
	}
}

/**
 * @description: 小规模写入数据
 * @param {string} path
 * @param {*[]string} data
 * @return {*}
 */
func WriteCsvLine(path string, data *[]string) {
	//OpenFile读取文件，不存在时则创建，使用追加模式
	File, err := os.OpenFile(path, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0666)
	if err != nil {
		log.Println("文件打开失败！")
	}
	defer File.Close()

	//创建写入接口
	WriterCsv := csv.NewWriter(File)
	err1 := WriterCsv.Write(*data)
	if err1 != nil {
		log.Println("WriterCsv写入文件失败")
	}
	WriterCsv.Flush() //刷新，不刷新是无法写入的
}
